package com.gy.hadoop.mr.demo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.Arrays;

/*

1.window本地运行MR程序：
    1.IDEA 本地执行，只需要maven配置大数据组件的相关jar包即可。
    2.需要：
            // 指定使用 本地执行
            conf.set("mapreduce.framework.name", "local")
            //指定本次 MapReduce程序中 所运行主类
            job.setJarByClass(xxx.class);
    3.不需要 job.setJar 和 System.setProperty("HADOOP_USER_NAME", "root")
    4.resources目录中 不需要 core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml

2.windwo推送MR程序到linux集群运行
    1.IDEA 首先需要maven配置大数据组件的相关jar包。
    2.需要：
            // 指定使用 远程yarn框架执行
            conf.set("mapreduce.framework.name", "yarn")
            // 如果搭建了 hadoop HA 的话，需要指定 活动状态下的 namenode
            conf.set("fs.defaultFS", "hdfs://node1:8020");
            // 如果要从windows系统中运行这个job提交客户端的程序，则需要加这个跨平台提交的参数
            // 配置使用跨平台提交任务，Windows开发者需要设置跨平台
            conf.setBoolean("mapreduce.app-submission.cross-platform", true);
            //通过这种方式设置java客户端访问hdfs的身份
            System.setProperty("HADOOP_USER_NAME", "root");
            // 指定jar文件，该jar文件为mapreduce程序打包后的jar文件，需要是可执行jar包，文件路径可以是本地文件路径或hdfs路径
            job.setJar("D:\\daima\\项目名\\target\\项目名-1.0-SNAPSHOT.jar");
            //指定本次 MapReduce程序中 所运行主类
            job.setJarByClass(xxx.class);
    3.resources目录中 需要 core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml


3.linux集群运行MR程序
    1.需要：
            // 指定使用 远程yarn框架执行
            conf.set("mapreduce.framework.name", "yarn")
            // 如果搭建了 hadoop HA 的话，需要指定 活动状态下的 namenode
            conf.set("fs.defaultFS", "hdfs://node1:8020");
            // 如果要从windows系统中运行这个job提交客户端的程序，则需要加这个跨平台提交的参数
            //通过这种方式设置java客户端访问hdfs的身份
            System.setProperty("HADOOP_USER_NAME", "root");
            //指定本次 MapReduce程序中 所运行主类
            job.setJarByClass(xxx.class);
    2.resources目录中 需要 core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml
      可在CDH的页面中点击下载hdfs、yarn的配置文件
 */
/**
 * 1/描述mr的shuffle全流程
 * 2/搭建maven工程，统计词频，提交集群运行，查看结果
 * 3/利用搜狗数据，找出所有独立的uid并写入hdfs
 * 4/利用搜狗数据，找出所有独立的uid出现次数，并写入hdfs，并要求使用map端的combine操作
 * 5/对mr数据倾斜，如何解决
 */

/*
shuffle流程
    1/框架调用map数据k,v对，写入k,v,p
    2/map写入环形缓冲区，默认缓冲区100m,满了80%开始溢写，溢写和写入同时进行
    3/溢写线程在写入磁盘时会进行一个快速排序（按照分区）
    4/溢写的小文件会触发combine，写的的小文件很多（默认3个）会触发merge(也会有combine,按照分区merge)
    5/reduce开始拉取属于自己分区的map计算结果文件，拉取来会有一个merge（会触发combine），文件数据太大会落地磁盘，一边计算一边拉取
 */
public class SougouMR {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
//        conf.set("mapreduce.app-submission.cross-platform", "true");

        Job job = Job.getInstance(conf, args[0]);
        job.setJar("target/kkb-bigdata-1.0-SNAPSHOT.jar");


        conf.set("mapred.job.tracker", "local");
//        conf.set("fs.defaultFS", "hdfs://gyClusterHa:50070");


        FileInputFormat.setInputPaths(job, args[1]);

        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(new Path(args[2]))) {
            fs.delete(new Path(args[2]), true);
        }

        FileOutputFormat.setOutputPath(job, new Path(args[2]));


        job4(job);


        job.waitForCompletion(true);
    }

    /**
     * 利用搜狗数据，找出所有独立的uid并写入hdfs
     * sougou_job3 /data/input/sougou/sogou.1w.utf8  /data/output/mr/sougou/job3
     *
     * @param job
     */
    private static void job3(Job job) {
        job.setMapperClass(Job3Map.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

//        job.setCombinerClass(Job3Reduce.class);

        job.setReducerClass(Job3Reduce.class);
    }

    /**
     * 利用搜狗数据，找出所有独立的uid并写入hdfs
     * sougou_job4 /data/input/sougou/sogou.1w.utf8  /data/output/mr/sougou/job4
     *
     * @param job
     */
    private static void job4(Job job) {
        job.setMapperClass(Job3Map.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setCombinerClass(Job4Reduce.class);


        job.setReducerClass(Job4Reduce.class);
    }

    /*
    date            uid                           searchName searchOrder clickOrder   link
20111230000005	57375476989eea12893c0c3811607bcf	奇艺高清	1	1	http://www.qiyi.com/
20111230000005	66c5bb7774e31d0a22278249b26bc83a	凡人修仙传	3	1	http://www.booksky.org/BookDetail.aspx?BookID=1050804&Level=1
*/
    public static class Job3Map extends Mapper<LongWritable, Text, Text, Text> {

        private final Text mKey = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] words = value.toString().split("\t");
            mKey.set(words[1]);//uid
            context.write(mKey, value);
        }
    }

    /*
    57375476989eea12893c0c3811607bcf    [
        (20111230000005	57375476989eea12893c0c3811607bcf	奇艺高清	1	1	http://www.qiyi.com/),
        (20111230000005	66c5bb7774e31d0a22278249b26bc83a	凡人修仙传	3	1	http://www.booksky.org/BookDetail.aspx?BookID=1050804&Level=1)
    ]
     */
    public static class Job3Reduce extends Reducer<Text, Text, Text, Text> {

        private final Text rVal = new Text();

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            int count = 0;
            StringBuffer sb = new StringBuffer();
            for (Text value : values) {
                if (count > 0) return;
                count++;
                sb.append(value + "|||");
            }
            rVal.set(sb.toString());
            if (count == 1)
                context.write(key, rVal);
        }
    }


    /*
57375476989eea12893c0c3811607bcf    [
    (20111230000005	57375476989eea12893c0c3811607bcf	奇艺高清	1	1	http://www.qiyi.com/),
    (20111230000005	66c5bb7774e31d0a22278249b26bc83a	凡人修仙传	3	1	http://www.booksky.org/BookDetail.aspx?BookID=1050804&Level=1)
]
 */
    public static class Job4Reduce extends Reducer<Text, Text, Text, Text> {

        private final Text rVal = new Text();

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            int count = 0;
            StringBuffer sb = new StringBuffer();
            for (Text value : values) {
                count++;
                sb.append(value + "|");
            }
            rVal.set(sb.insert(0, count + "::").toString());
            context.write(key, rVal);
        }
    }

    static class Productor implements Comparable<Productor> {
        private String name;
        private Double price;
        private Integer count;
        private String area;

        public Productor() {
        }

        public Productor(String name, Double price, Integer count, String area) {
            this.name = name;
            this.price = price;
            this.count = count;
            this.area = area;
        }


        public int compareTo(Productor p) {
            double tmp = this.price - p.price;
            return tmp == 0D ? 0 : tmp > 0 ? 1 : -1;
        }

        String show() {
            final StringBuffer sb = new StringBuffer("Productor{");
            sb.append("name='").append(name).append('\'');
            sb.append(", price='").append(price).append('\'');
            sb.append(", count='").append(count).append('\'');
            sb.append(", area='").append(area).append('\'');
            sb.append('}');
            return sb.toString();
        }

        void job4() {
            Productor[] ps = new Productor[]{
                    new Productor("苹果", 22.2D, 11, "武汉"),
                    new Productor("香蕉", 1.1D, 2, "黄冈"),
                    new Productor("橘子", 5.212D, 3, "武穴")
            };

            for (int i = 0; i < ps.length; i++)
                ps[i].show();
            System.out.println("==============================");


            double sum = 0D;
            for (int i = 0; i < ps.length; i++)
                sum += ps[i].price;
            System.out.println(String.format("总价格：%s", sum));
            System.out.println("==============================");


            Arrays.sort(ps);
            System.out.println(String.format("价格最高的商品：%s", ps[ps.length - 1]));
            System.out.println("==============================");
        }


    }
}
